"""
@File    :   service.py
@Time    :   2024/02/29 11:21:23
@Author  :   Qrj
@Email   :   923207736@qq.com
@description   :   token的curd相关方法类
"""

from datetime import datetime, timedelta

from redis.asyncio import Redis
from sqlalchemy import and_, delete, func, insert, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.business.admin.system.oauth2.client.service import Oauth2ClientService
from src.schemas.validate import (
    OrderBySearchValidateSchema,
    PaginationSearchValidateSchema,
)
from src.utils.cashews import token_cache
from src.utils.datetime import DatetimeUtil

from .boost_task import (
    create_double_token_clean_user_logged_tokens_zset_handler,
    delete_double_tokens_clean_user_logged_token_zset_handler,
    refresh_access_token_clean_user_logged_token_zset_handler,
)
from .exceptions import access_token_expired_exception, refresh_token_expired_exception
from .models import AccessToken, RefreshToken
from .schemas.out import (
    AccessTokenCompleteOutSchema,
    DoubleTokenCompleteOutSchema,
    ListAccessTokenCompleteOutSchema,
    RefreshTokenCompleteOutSchema,
)
from .schemas.validate import (
    AccessTokenCreateValidateSchema,
    AccessTokenListSearchValidateSchema,
    DoubleTokenCreateValidateSchema,
    RefreshTokenCreateValidateSchema,
)
from .typings import RefreshTokenAndUserIdMappingsT


class TokenService:
    """Token相关Curd方法类"""

    @staticmethod
    def generate_access_token_cashews_key(access_token: str) -> str:
        """生成访问令牌的cashews的key

        Args:
            access_token (str): 访问令牌

        Returns:
            str: 访问令牌的cashews的key
        """
        return f"access_token:{access_token}"

    @staticmethod
    def generate_refresh_token_cashews_key(refresh_token: str) -> str:
        """生成刷新令牌的cashews的key

        Args:
            refresh_token (str): 刷新令牌

        Returns:
            str: 刷新令牌的cashews的key
        """
        return f"refresh_token:{refresh_token}"

    @staticmethod
    def generate_user_logged_tokens_zset_name(user_id: int) -> str:
        """生成用户在Redis中对应的已登录Token的zset的名称

        Args:
            user_id (int): 用户ID

        Returns:
            str: 用户在Redis中对应的已登录Token的zset的名称
        """
        return f"user_{user_id}_logged_tokens_zset"

    @staticmethod
    def generate_user_logged_tokens_zset_mapping(
        access_token: str,
        refresh_token: str,
        refresh_token_expiration_datetime: datetime,
    ) -> dict[str, int]:
        """生成用户在Redis中对应的已登录Token的zset的member与scope的mapping

        Args:
            access_token (str): 访问令牌
            refresh_token (str): 刷新令牌
            refresh_token_expiration_datetime (datetime): 刷新令牌的过期时间

        Returns:
            dict[str, int]: mapping
        """
        member = f"{access_token}:{refresh_token}"
        score: int = DatetimeUtil.to_timestamp(
            refresh_token_expiration_datetime,
            level="millisecond",
            rounding=True,
        )
        return {member: score}

    @staticmethod
    async def get_access_token(
        access_token: str,
        db_session: AsyncSession | None = None,
        raise_not_exists_exception: bool = True,
    ) -> AccessTokenCompleteOutSchema | None:
        """获取access_token信息实例

        Args:
            access_token (str): 访问令牌
            db_session (AsyncSession | None, optional): 数据库连接池, 如果不需要从sql中查找则可以不传
            raise_not_exists_exception(bool, optional): 没找到的话是否直接报错

        Returns:
            AccessTokenCompleteOutSchema | None: 访问令牌数据
        """
        # 构建访问令牌的redis缓存的key
        access_token_key = TokenService.generate_access_token_cashews_key(access_token)
        # 先从缓存中取
        access_token_info_cache = await token_cache.get(access_token_key)
        if access_token_info_cache is None:
            # 缓存中未找到时
            if db_session is not None:
                # 缓存中没有且有传入数据库连接池时, 去sql中查找
                search_sql = select(
                    *AccessTokenCompleteOutSchema.convert_to_sql_model_field(
                        AccessToken
                    )
                ).where(AccessToken.access_token == access_token)
                async with db_session.begin():
                    search_res = await db_session.execute(search_sql)
                result = search_res.mappings().first()
                if result is not None:
                    # sql中找到时
                    # 将sql结果转换为pydantic的序列化模型
                    access_token_info = AccessTokenCompleteOutSchema.model_validate(
                        result
                    )
                    # 计算缓存过期时间(秒)
                    cache_expire_seconds = (
                        DatetimeUtil.get_diff_seconds_from_current_time(
                            access_token_info.expiration_datetime
                        )
                    )
                    # 当过期时间时间戳大于当前时间戳时, 将sql中的数据存入缓存
                    if cache_expire_seconds > 0:
                        await token_cache.set(
                            access_token_key,
                            access_token_info,
                            expire=cache_expire_seconds,
                        )
                    return access_token_info
                else:
                    # sql中未找到时
                    if raise_not_exists_exception:
                        raise access_token_expired_exception
                    return
            else:
                # 缓存中没有并且没有传入数据库连接池时
                if raise_not_exists_exception:
                    raise access_token_expired_exception
                return
        else:
            # 缓存中找到时
            # 直接返回缓存
            return access_token_info_cache

    @staticmethod
    async def get_refresh_token(
        refresh_token: str,
        db_session: AsyncSession | None = None,
        raise_not_exists_exception: bool = True,
    ) -> RefreshTokenCompleteOutSchema | None:
        """获取refresh_token信息实例

        Args:
            refresh_token (str): 刷新令牌
            db_session (AsyncSession | None, optional): 数据库连接池, 如果不需要从sql中查找则可以不传
            raise_not_exists_exception(bool, optional): 没找到的话是否直接报错

        Returns:
            RefreshTokenCompleteOutSchema | None: 刷新令牌数据
        """
        # 构建redis的key
        refresh_token_key = TokenService.generate_refresh_token_cashews_key(
            refresh_token
        )
        # 先从缓存中取
        refresh_token_info_cache = await token_cache.get(refresh_token_key)
        if refresh_token_info_cache is None:
            # 缓存中未找到时
            if db_session is not None:
                # 缓存中没有且有传入数据库连接池时, 去sql中查找
                search_sql = select(
                    *RefreshTokenCompleteOutSchema.convert_to_sql_model_field(
                        RefreshToken
                    )
                ).where(RefreshToken.refresh_token == refresh_token)
                async with db_session.begin():
                    search_res = await db_session.execute(search_sql)
                refresh_token_info = search_res.mappings().first()
                if refresh_token_info is not None:
                    # sql中找到时
                    # 将sql结果转换为pydantic的序列化模型
                    refresh_token_info = RefreshTokenCompleteOutSchema.model_validate(
                        refresh_token_info
                    )
                    # 计算缓存过期时间(秒)
                    cache_expire_seconds = (
                        DatetimeUtil.get_diff_seconds_from_current_time(
                            refresh_token_info.expiration_datetime
                        )
                    )
                    # 当过期时间时间戳大于当前时间戳时, 将sql中的数据存入缓存
                    if cache_expire_seconds > 0:
                        await token_cache.set(
                            refresh_token_key,
                            refresh_token_info,
                            expire=cache_expire_seconds,
                        )
                    return refresh_token_info
                else:
                    # sql中未找到时
                    if raise_not_exists_exception:
                        raise refresh_token_expired_exception
                    return
            else:
                # 缓存中没有并且没有传入数据库连接池时
                if raise_not_exists_exception:
                    raise refresh_token_expired_exception
                return
        else:
            # 缓存中找到时
            # 直接返回缓存
            return refresh_token_info_cache

    @staticmethod
    async def list_search_access_token(
        db_session: AsyncSession,
        tenant_id: int,
        search_schema: AccessTokenListSearchValidateSchema,
        pagination_schema: PaginationSearchValidateSchema,
        order_schema: OrderBySearchValidateSchema,
        need_select_total: bool = True,
    ) -> tuple[list[AccessTokenCompleteOutSchema], int | None]:
        """完整信息的访问令牌列表查询

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id (int): 租户ID
            search_schema (AccessTokenListSearchValidateSchema): 查询参数
            pagination_schema (PaginationSearchValidateSchema): 分页参数
            order_schema (OrderBySearchValidateSchema): 排序参数
            need_select_total (bool, optional): 是否需要查询总数

        Returns:
            tuple[list[AccessTokenCompleteOutSchema], int | None]: 查询结果、查询总数
        """
        # 构建排序字段
        order_by_field_sql = order_schema.convert_to_order_by_sql(
            search_schema, AccessToken
        )
        # 构建查询sql
        list_search_sql = (
            select(
                *AccessTokenCompleteOutSchema.convert_to_sql_model_field(AccessToken)
            )
            .where(
                and_(*search_schema.convert_to_where_sql(AccessToken)),
                AccessToken.tenant_id == tenant_id,
            )
            .order_by(order_by_field_sql)
            .offset(pagination_schema.offset)
            .limit(pagination_schema.limit)
        )
        # 开始查询
        async with db_session.begin():
            list_search_res = await db_session.execute(list_search_sql)
        result = list_search_res.mappings().all()
        # 根据need_select_total参数来确定是否需要查询总数
        if need_select_total:
            total_sql = (
                select(func.count())
                .select_from(AccessToken)
                .where(
                    and_(*search_schema.convert_to_where_sql(AccessToken)),
                    AccessToken.tenant_id == tenant_id,
                )
            )
            async with db_session.begin():
                total_res = await db_session.execute(total_sql)
            total = total_res.scalar()
        else:
            total = None
        # 统一转为pydantic列表模型
        return ListAccessTokenCompleteOutSchema.validate_python(result), total

    @token_cache.transaction()
    @staticmethod
    async def create_double_token(
        db_session: AsyncSession,
        token_redis_pool: Redis,
        create_schema: DoubleTokenCreateValidateSchema,
    ) -> DoubleTokenCompleteOutSchema:
        """创建双Token

        Args:
            db_session (AsyncSession): 数据库连接池
            token_redis_pool: token的redis连接池
            create_schema: 创建所需的信息

        Returns:
            DoubleTokenCompleteOutSchema: 双Token信息
        """
        # 进行Token数据验证, 并生成双token数据
        # 获取创建时间
        create_datetime = datetime.now()
        # 获取过期时间
        access_token_expiration_datetime = create_datetime + timedelta(
            seconds=create_schema.access_token_expiration_seconds
        )
        refresh_token_expiration_datetime = create_datetime + timedelta(
            seconds=create_schema.refresh_token_expiration_seconds
        )
        # 生成刷新令牌数据
        refresh_token_create_schema = RefreshTokenCreateValidateSchema(
            create_datetime=create_datetime,
            expiration_datetime=refresh_token_expiration_datetime,
            **create_schema.model_dump(),
        )
        # 生成访问令牌数据
        access_token_create_schema = AccessTokenCreateValidateSchema(
            create_datetime=create_datetime,
            expiration_datetime=access_token_expiration_datetime,
            **create_schema.model_dump(),
        )
        # 向sql中插入双token数据(开启sql事务)
        async with db_session.begin():
            # 先创建refresh_token
            refresh_token_create_sql = insert(RefreshToken).values(
                **refresh_token_create_schema.model_dump()
            )
            refresh_token_created_res = await db_session.execute(
                refresh_token_create_sql
            )
            # 再创建access_token
            access_token_create_sql = insert(AccessToken).values(
                **access_token_create_schema.model_dump()
            )
            access_token_created_res = await db_session.execute(access_token_create_sql)
            # 设置双Token的缓存
            # 将数据转换为pydantic的序列化模型(为了统一缓存中token数据类型)
            refresh_token_out_info = RefreshTokenCompleteOutSchema(
                **refresh_token_create_schema.model_dump(),
                id=refresh_token_created_res.inserted_primary_key_rows[0][0],
            )
            access_token_out_info = AccessTokenCompleteOutSchema(
                **access_token_create_schema.model_dump(),
                id=access_token_created_res.inserted_primary_key_rows[0][0],
            )
            await token_cache.set(
                TokenService.generate_refresh_token_cashews_key(
                    refresh_token_create_schema.refresh_token
                ),
                refresh_token_out_info,
                expire=create_schema.refresh_token_expiration_seconds,
            )
            await token_cache.set(
                TokenService.generate_access_token_cashews_key(
                    access_token_create_schema.access_token
                ),
                access_token_out_info,
                expire=create_schema.access_token_expiration_seconds,
            )
            # 判断是否需要对当前登录用户的已登录Token的zset进行维护
            if create_schema.user_id and create_schema.occupation_of_login_quota:
                # 维护当前登录用户的已登录Token的zset
                # member为"access_token:refresh_token", score为refresh_token过期的时间戳
                zset_name = TokenService.generate_user_logged_tokens_zset_name(
                    create_schema.user_id
                )
                member_score_mapping = (
                    TokenService.generate_user_logged_tokens_zset_mapping(
                        access_token_create_schema.access_token,
                        refresh_token_create_schema.refresh_token,
                        refresh_token_create_schema.expiration_datetime,
                    )
                )
                async with token_redis_pool.pipeline() as pipe:
                    # 将双token信息写入对应user_id的已登录Token的zset中
                    await pipe.zadd(zset_name, member_score_mapping)
                    # 进行当前用户最大登录数量检测zcount
                    await pipe.zcount(zset_name, 0, float("+inf"))
                    _, logged_token_count = await pipe.execute()
                # 当前用户超出最大同时登录数量时, 使用异步任务队列进行zset清理
                if (
                    logged_token_count
                    > create_schema.user_number_of_simultaneous_logins
                ):
                    # 异步任务调用
                    create_double_token_clean_user_logged_tokens_zset_handler.push(
                        create_schema.user_id,
                        create_schema.user_number_of_simultaneous_logins,
                    )
        return DoubleTokenCompleteOutSchema(
            access_token=access_token_out_info, refresh_token=refresh_token_out_info
        )

    @staticmethod
    async def create_double_token_clean_user_logged_tokens_zset(
        db_session: AsyncSession,
        token_redis_pool: Redis,
        user_id: int,
        user_number_of_simultaneous_logins: int,
    ):
        """登录(创建双Tokne)时清理用户的已登录双token的zset(一般使用异步任务队列调用)

        Args:
            db_session (AsyncSession): 数据库连接池
            token_redis_pool (Redis): token的redis连接池
            user_id (int): 用户ID
            user_number_of_simultaneous_logins (int): 用户最多可同时登录数量
        """
        # 生成对应用户的zset名称
        zset_name = TokenService.generate_user_logged_tokens_zset_name(user_id)
        # 获取zset当前元素数量
        zset_count = await token_redis_pool.zcount(zset_name, 0, float("+inf"))
        if zset_count > user_number_of_simultaneous_logins:
            # 当超出用户最多可同时登录数量时, 删除过期时间最短的x个元素
            poped_tokens = await token_redis_pool.zpopmin(
                zset_name, zset_count - user_number_of_simultaneous_logins
            )
            # 批量删除双Token的缓存与sql
            poped_tokens = [poped_token for poped_token, _ in poped_tokens]
            await TokenService.delete_double_tokens(db_session, poped_tokens)

    @token_cache.transaction()
    @staticmethod
    async def refresh_access_token(
        db_session: AsyncSession,
        token_redis_pool: Redis,
        refresh_token: str,
    ) -> DoubleTokenCompleteOutSchema:
        """使用refresh_token刷新access_token

        Args:
            db_session (AsyncSession): 数据库连接池
            token_redis_pool (Redis): redis连接池
            refresh_token (str): 刷新token

        Returns:
            AccessTokenCompleteOutSchema: 新生成的访问令牌信息
        """
        # 获取刷新令牌信息,过期报错， 没过期则返回
        refresh_token_info: RefreshTokenCompleteOutSchema = (
            await TokenService.get_refresh_token(refresh_token, db_session=db_session)
        )
        # 获取Oauth2客户端信息
        oauth2_client_info = await Oauth2ClientService.get_oauth2_client_use_id(
            db_session,
            refresh_token_info.tenant_id,
            refresh_token_info.oauth2_client_id,
        )
        # 开启sql事务, 搭配cashews的事务
        async with db_session.begin():
            # 生成新的访问令牌相关数据
            # 创建时间
            create_datetime = datetime.now()
            # 过期时间
            expiration_datetime = create_datetime + timedelta(
                seconds=oauth2_client_info.access_token_expiration_seconds
            )
            # 创建新的访问令牌
            access_token_create_schema = AccessTokenCreateValidateSchema(
                tenant_id=refresh_token_info.tenant_id,
                tenant_is_system=refresh_token_info.tenant_is_system,
                tenant_pakcage_id=refresh_token_info.tenant_pakcage_id,
                create_datetime=create_datetime,
                expiration_datetime=expiration_datetime,
                oauth2_client_id=oauth2_client_info.id,
                oauth2_grant_type=refresh_token_info.oauth2_grant_type,
                oauth2_scopes=refresh_token_info.oauth2_scopes,
                user_id=refresh_token_info.user_id,
                username=refresh_token_info.username,
                user_is_admin=refresh_token_info.user_is_admin,
                token_type=refresh_token_info.token_type,
                refresh_token=refresh_token_info.refresh_token,
            )
            # 写入sql
            access_token_create_sql = insert(AccessToken).values(
                **access_token_create_schema.model_dump()
            )
            access_token_created_res = await db_session.execute(access_token_create_sql)
            # 写入缓存
            # 转为序列化模型
            access_token_info = AccessTokenCompleteOutSchema(
                **access_token_create_schema.model_dump(),
                id=access_token_created_res.inserted_primary_key_rows[0][0],
            )
            await token_cache.set(
                TokenService.generate_access_token_cashews_key(
                    access_token_info.access_token
                ),
                access_token_info,
                expire=oauth2_client_info.access_token_expiration_seconds,
            )
            # 维护当前登录用户的已登录Token的zset
            if access_token_info.user_id:
                # member为"access_token:refresh_token", score为refresh_token过期的时间戳
                zset_name = TokenService.generate_user_logged_tokens_zset_name(
                    access_token_info.user_id
                )
                member_score_mapping = (
                    TokenService.generate_user_logged_tokens_zset_mapping(
                        access_token_info.access_token,
                        refresh_token_info.refresh_token,
                        refresh_token_info.expiration_datetime,
                    )
                )
                # 将双token信息写入对应user_id的已登录Token的zset中
                await token_redis_pool.zadd(zset_name, member_score_mapping)
                # 使用异步任务队列进行zset清理和旧的access_token的缓存和sql清理
                refresh_access_token_clean_user_logged_token_zset_handler.push(
                    access_token_info.user_id,
                    access_token_info.access_token,
                    access_token_info.refresh_token,
                )
        return DoubleTokenCompleteOutSchema(
            access_token=access_token_info, refresh_token=refresh_token_info
        )

    @token_cache.transaction()
    @staticmethod
    async def refresh_access_token_clean_user_logged_token_zset(
        db_session: AsyncSession,
        token_redis_pool: Redis,
        user_id: int,
        new_access_token: str,
        refresh_token: str,
    ):
        """使用刷新令牌刷新访问令牌时清理用户的已登录双token的zset(一般使用异步任务队列调用)

        Args:
            db_session (AsyncSession): 数据库连接池
            token_redis_pool (Redis): redis连接池
            user_id (int): 用户ID
            new_access_token (str): 使用refresh_token刷新出的新的access_token
            refresh_token (str): 刷新令牌
        """
        # 生成对应用户的zset名称
        zset_name = TokenService.generate_user_logged_tokens_zset_name(user_id)
        # 扫描出当前zset中refresh_token对应的双token字符
        match_str = f"*:{refresh_token}"
        # 删除除了当前的双token以外的refresh_token对应的其他双Token
        not_delete_double_token = f"{new_access_token}:{refresh_token}"
        need_delete_double_tokens = [
            i
            async for i, _ in token_redis_pool.zscan_iter(zset_name, match=match_str)
            if i != not_delete_double_token
        ]
        await token_redis_pool.zrem(zset_name, *need_delete_double_tokens)
        # 删除access_token的sql和缓存
        if need_delete_double_tokens:
            # 生成access_token列表
            need_delete_access_tokens = [
                i.split(":")[0] for i in need_delete_double_tokens
            ]
            del_access_token_sql = delete(AccessToken).where(
                AccessToken.access_token.in_(need_delete_access_tokens)
            )
            # 生成缓存的key
            need_delete_access_token_cache_keys = [
                TokenService.generate_access_token_cashews_key(i)
                for i in need_delete_access_tokens
            ]
            async with db_session.begin():
                await token_cache.delete_many(*need_delete_access_token_cache_keys)
                await db_session.execute(del_access_token_sql)

    @token_cache.transaction()
    @staticmethod
    async def delete_double_tokens(
        db_session: AsyncSession,
        double_tokens: list[str],
    ):
        """只删除双Token的缓存与sql
        由异步任务队列消费者调用(也等于是在异步任务中调用)

        Args:
            db_session (AsyncSession): 数据库连接池
            double_tokens (list[str]): 双Token, 格式为{access_token}:{refresh_token}
        """
        if double_tokens:
            # 拆分出来的token列表
            access_token_list = []
            refresh_token_list = []
            # 批量删除双Token缓存
            access_token_keys = []
            refresh_token_keys = []
            # 填充token列表与需删除的缓存的key的列表
            for double_token in double_tokens:
                access_token, refresh_token = double_token.split(":")
                access_token_keys.append(
                    TokenService.generate_access_token_cashews_key(access_token)
                )
                access_token_list.append(access_token)
                refresh_token_keys.append(
                    TokenService.generate_refresh_token_cashews_key(refresh_token)
                )
                refresh_token_list.append(refresh_token)
            # 批量删除访问令牌的sql
            del_access_token_sql = delete(AccessToken).where(
                or_(
                    *[AccessToken.refresh_token == i for i in refresh_token_list],
                    *[AccessToken.access_token == i for i in access_token_list],
                )
            )
            del_refresh_token_sql = delete(RefreshToken).where(
                or_(*[RefreshToken.refresh_token == i for i in refresh_token_list])
            )
            async with db_session.begin():
                # 删除缓存
                await token_cache.delete_many(*access_token_keys, *refresh_token_keys)
                # 删除sql
                await db_session.execute(del_access_token_sql)
                await db_session.execute(del_refresh_token_sql)

    @staticmethod
    async def delete_double_tokens_clean_user_logged_token_zset(
        token_redis_pool: Redis,
        refresh_token_and_user_id_mappings: RefreshTokenAndUserIdMappingsT,
    ):
        """删除双token时清理用户的已登录双token的zset(异步任务队列和同步调用)

        Args:
            token_redis_pool (Redis): redis连接池
            refresh_token_and_user_id_mappings (RefreshTokenAndUserIdMappingsT): 刷新token及其对应的用户id的映射列表
        """
        for refresh_tokens_and_user_id in refresh_token_and_user_id_mappings:
            zset_name = TokenService.generate_user_logged_tokens_zset_name(
                refresh_tokens_and_user_id["user_id"]
            )
            need_delete_double_tokens = [
                i
                async for i, _ in token_redis_pool.zscan_iter(
                    zset_name,
                    match=f"*:{refresh_tokens_and_user_id['refresh_token']}",
                )
            ]
            await token_redis_pool.zrem(zset_name, *need_delete_double_tokens)

    @token_cache.transaction()
    @staticmethod
    async def delete_expired_double_tokens(
        db_session: AsyncSession,
        token_redis_pool: Redis,
    ):
        """删除过期的双Token(一般使用异步任务队列的定时任务调用)

        Args:
            db_session (AsyncSession): 数据库连接池
            token_redis_pool (Redis): redis连接池
        """
        async with db_session.begin():
            # 获取当前时间
            now_datetime = datetime.now()
            # 先查询出过期的refresh_token及其对应的user_id(用于清理最大同时登录zset)
            select_expired_refresh_tokens_sql = select(
                RefreshToken.refresh_token, RefreshToken.user_id
            ).where(RefreshToken.expiration_datetime <= now_datetime)
            expired_refresh_tokens = await db_session.execute(
                select_expired_refresh_tokens_sql
            )
            expired_refresh_tokens_and_user_id = expired_refresh_tokens.mappings().all()
            # 筛选出过期的refresh_token列表
            expired_refresh_tokens = [
                i["refresh_token"] for i in expired_refresh_tokens_and_user_id
            ]
            # 删除查询到的过期的refresh_token
            if expired_refresh_tokens:
                delete_expired_refresh_tokens_sql = delete(RefreshToken).where(
                    or_(
                        *[
                            RefreshToken.refresh_token == i
                            for i in expired_refresh_tokens
                        ]
                    )
                )
                await db_session.execute(delete_expired_refresh_tokens_sql)
            # 再查询过期的refresh_token关联的access_token和过期的access_token
            select_expired_access_tokens_sql = select(AccessToken.access_token).where(
                or_(
                    *[AccessToken.refresh_token == i for i in expired_refresh_tokens],
                    AccessToken.expiration_datetime <= now_datetime,
                )
            )
            expired_access_tokens = await db_session.execute(
                select_expired_access_tokens_sql
            )
            expired_access_tokens = expired_access_tokens.scalars().all()
            if expired_access_tokens:
                # 删除查询到的过期的access_token
                delete_expired_access_tokens_sql = delete(AccessToken).where(
                    or_(*[AccessToken.access_token == i for i in expired_access_tokens])
                )
                await db_session.execute(delete_expired_access_tokens_sql)
            if expired_refresh_tokens or expired_access_tokens:
                # 删除双Token的缓存
                await token_cache.delete_many(
                    *[
                        TokenService.generate_refresh_token_cashews_key(i)
                        for i in expired_refresh_tokens
                    ],
                    *[
                        TokenService.generate_access_token_cashews_key(i)
                        for i in expired_access_tokens
                    ],
                )
            if expired_access_tokens:
                # 清理用户的管理最大同时登录数量的zset
                await TokenService.delete_double_tokens_clean_user_logged_token_zset(
                    token_redis_pool, expired_refresh_tokens_and_user_id
                )

    @staticmethod
    async def logout_double_tokens(
        db_session: AsyncSession,
        tenant_id: int,
        refresh_tokens: list[str],
    ):
        """登出或强退双Token

        Args:
            db_session (AsyncSession): 数据库连接池
            tenant_id
            refresh_tokens (list[str]): 刷新令牌列表
        """
        # 查出需要logout的refresh_token和access_token的信息
        select_logout_double_tokens_sql = select(
            AccessToken.refresh_token, AccessToken.user_id, AccessToken.access_token
        ).where(
            or_(*[AccessToken.refresh_token == i for i in refresh_tokens]),
            AccessToken.tenant_id == tenant_id,
        )
        async with db_session.begin():
            logout_double_tokens = await db_session.execute(
                select_logout_double_tokens_sql
            )
        logout_double_tokens_mappings = logout_double_tokens.mappings().all()
        # 有查到才进行删除和后续操作
        if logout_double_tokens_mappings:
            logout_double_tokens_mappings = [
                dict(i) for i in logout_double_tokens_mappings
            ]
            # 构建需删除的双token列表
            need_delete_double_tokens = [
                f"{i['access_token']}:{i['refresh_token']}"
                for i in logout_double_tokens_mappings
            ]
            # 删除sql和缓存(直接调用)
            await TokenService.delete_double_tokens(
                db_session, need_delete_double_tokens
            )
            # 清理用户的已登录双token的zset(异步任务队列进行消费)
            delete_double_tokens_clean_user_logged_token_zset_handler.push(
                logout_double_tokens_mappings
            )
